package org.senseframework.support.orm.hbase.datasource;

import org.apache.commons.collections.MapUtils;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.senseframework.support.orm.hbase.annotation.HBaseTable;
import org.senseframework.support.orm.hbase.definition.HBaseColumnDefinition;
import org.senseframework.support.orm.hbase.definition.HBaseEntityDefinition;
import org.senseframework.support.orm.hbase.utils.HBaseEntityUtils;
import org.springframework.context.ApplicationContext;

import java.util.Map;

/**
 * @author wangqian
 * @date 18/4/14 下午3:13
 */
public class AutoCreateTable {

    private HBaseAdmin admin;

    public AutoCreateTable(HBaseAdmin admin) {
        this.admin = admin;
    }

    public void start(ApplicationContext applicationContext) throws Exception {
        Map<String, Object> beanMap = applicationContext.getBeansWithAnnotation(HBaseTable.class);
        if (MapUtils.isNotEmpty(beanMap)) {
            for (Object object : beanMap.values()) {
                HBaseEntityDefinition entityDefinition = HBaseEntityUtils.findEntityDefinition(object.getClass(), true);
                if (admin.tableExists(entityDefinition.getTable())) {
                    //表存在，比较family
                    HTableDescriptor tableDescriptor = admin.getTableDescriptor(entityDefinition.tableBytes());
                    for (HBaseColumnDefinition columnDefinition : entityDefinition.getColumnDefinitions().values()) {
                        if (!tableDescriptor.hasFamily(columnDefinition.familyBytes())) {
                            HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnDefinition.familyBytes());
                            admin.addColumn(entityDefinition.getTable(), columnDescriptor);
                        }
                    }
                } else {
                    //表不存在
                    HTableDescriptor tableDescriptor = new HTableDescriptor(entityDefinition.getTable());
                    for (HBaseColumnDefinition columnDefinition : entityDefinition.getColumnDefinitions().values()) {
                        if (!tableDescriptor.hasFamily(columnDefinition.familyBytes())) {
                            HColumnDescriptor columnDescriptor = new HColumnDescriptor(columnDefinition.familyBytes());
                            tableDescriptor.addFamily(columnDescriptor);
                        }
                    }
                    admin.createTable(tableDescriptor);
                }
            }
        }
    }
}
